package com.ideaaedi.springcloud.jd.commonspring.sharding.support;

import com.alibaba.fastjson2.JSON;
import com.google.common.collect.Lists;
import com.ideaaedi.springcloud.jd.commonspring.sharding.bo.UpdateActualDataNodeBO;
import com.ideaaedi.springcloud.jd.commonspring.sharding.properties.DatasourceArgs;
import com.ideaaedi.springcloud.jd.commonspring.sharding.properties.RulesShardingProperties;
import com.ideaaedi.springcloud.jd.commonspring.util.DatabaseUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.lang.NonNull;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 本地缓存实现
 *
 * @author JustryDeng
 * @since 2021.0.5.A
 */
@Slf4j
public class LocalCacheShardingTableSupport implements ShardingTableSupport, SmartInitializingSingleton {
    
    @Resource
    DatasourceArgs datasourceArgs;
    
    @Resource
    RefreshActualDataNodesSupport refreshActualDataNodesSupport;
    
    @Resource
    RulesShardingProperties rulesShardingProperties;
    
    /**
     * 缓存实现
     * <br />
     * key - 逻辑表名
     * value - 真实表名集合
     */
    private static final ConcurrentHashMap<String, Set<String>> LOGIC_REAL_TABLE_NAME_CACHE = new ConcurrentHashMap<>(64);
    
    /**
     * 表ddl
     * <br />
     * key - 逻辑表名
     * value - 对应的ddl
     */
    public static final ConcurrentHashMap<String, String> TABLE_DDL_MAP = new ConcurrentHashMap<>(64);
    
    static {
        // msg_exec_log 表
        TABLE_DDL_MAP.put("msg_exec_log", "CREATE TABLE IF NOT EXISTS `" + "%s" + "` (\n"
                + "     `id` bigint(20) UNSIGNED NOT NULL COMMENT 'id',\n"
                + "     `tenant` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT "
                + "'租户（以特定的dept_path作为租户值）',\n"
                + "     `yyyy_mm_dd` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT "
                + "'消息的年月日',\n"
                + "     `msg_id` bigint(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT '消息 id',\n"
                + "     `curr_expect_exec_time` datetime NOT NULL COMMENT '触发本次执行的期望执行时间',\n"
                + "     `actual_exec_time` datetime NOT NULL COMMENT '实际执行时间',\n"
                + "     `consume_time` int(10) NOT NULL COMMENT '耗时（毫秒）',\n"
                + "     `exec_if_success` tinyint(1) NULL DEFAULT NULL COMMENT '执行是否成功（仅代表执行这个动作是否成功；0-失败；1-成功）',\n"
                + "     `exec_result` varchar(500) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL "
                + "COMMENT '执行结果',\n"
                + "     `status` tinyint(1) NOT NULL DEFAULT 1 COMMENT '数据状态(0-无效；1-有效)',\n"
                + "     `created_by` bigint(20) NULL DEFAULT NULL COMMENT '创建人',\n"
                + "     `created_at` datetime NULL DEFAULT NULL COMMENT '创建时间',\n"
                + "     `updated_by` bigint(20) NULL DEFAULT NULL COMMENT '修改人',\n"
                + "     `updated_at` datetime NULL DEFAULT NULL COMMENT '修改时间',\n"
                + "     PRIMARY KEY (`id`) USING BTREE,\n"
                + "     INDEX `status_idx`(`status` ASC) USING BTREE,\n"
                + "     INDEX `tenant_idx`(`tenant` ASC) USING BTREE,\n"
                + "     INDEX `msg_id_idx`(`msg_id` ASC) USING BTREE,\n"
                + "     INDEX `exec_if_success_idx`(`exec_if_success` ASC) USING BTREE\n"
                + ") ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '消息执行日志' ROW_FORMAT = "
                + "DYNAMIC;");
        // msg_from_to 表
        TABLE_DDL_MAP.put("msg_from_to", "CREATE TABLE IF NOT EXISTS `" + "%s" + "` (\n"
                + "    `id` bigint(20) UNSIGNED NOT NULL COMMENT 'id',\n"
                + "    `tenant` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT "
                + "'租户（以特定的dept_path作为租户值）',\n"
                + "    `yyyy_mm_dd` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT"
                + " '年月日',\n"
                + "    `from_type` varchar(50) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '发送方类型',\n"
                + "    `from_user_id` bigint(20) NULL DEFAULT NULL COMMENT '发送方用户id',\n"
                + "    `to_type` varchar(50) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '接收放类型',\n"
                + "    `to_user_phone` bigint(20) NULL DEFAULT NULL COMMENT '接收方为：此用户的手机号',\n"
                + "    `to_phone` varchar(50) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT "
                + "'接收方：此电话号码',\n"
                + "    `state` tinyint(1) NULL DEFAULT NULL COMMENT "
                + "'交互状态（-1-已失效；0-待发送；1-已发送，发送成功；2-已发送，发送失败；3-已发送，结果未知）',\n"
                + "    `status` tinyint(1) NOT NULL DEFAULT 1 COMMENT '数据状态(0-无效；1-有效)',\n"
                + "    `created_by` bigint(20) NULL DEFAULT NULL COMMENT '创建人',\n"
                + "    `created_at` datetime NULL DEFAULT NULL COMMENT '创建时间',\n"
                + "    `updated_by` bigint(20) NULL DEFAULT NULL COMMENT '修改人',\n"
                + "    `updated_at` datetime NULL DEFAULT NULL COMMENT '修改时间',\n"
                + "    PRIMARY KEY (`id`) USING BTREE,\n"
                + "    INDEX `status_idx`(`status` ASC) USING BTREE,\n"
                + "    INDEX `from_type_idx`(`from_type` ASC) USING BTREE,\n"
                + "    INDEX `to_type_idx`(`to_type` ASC) USING BTREE,\n"
                + "    INDEX `state_idx`(`state` ASC) USING BTREE,\n"
                + "    INDEX `tenant_idx`(`tenant` ASC) USING BTREE\n"
                + ") ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '消息交互方' ROW_FORMAT = "
                + "DYNAMIC;");
        // msg_message 表
        TABLE_DDL_MAP.put("msg_message", "CREATE TABLE IF NOT EXISTS `" + "%s" + "`  (\n"
                + "    `id` bigint(20) UNSIGNED NOT NULL COMMENT 'id',\n"
                + "    `tenant` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT "
                + "'租户（以特定的dept_path作为租户值）',\n"
                + "    `yyyy_mm_dd` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT"
                + " '年月日',\n"
                + "    `root_id` bigint(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT '根消息 id（0表示无）',\n"
                + "    `parent_id` bigint(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT '父消息 id（0表示无）',\n"
                + "    `msg_type` tinyint(1) NOT NULL COMMENT '消息类型(0-即时消息; 1-定时消息)',\n"
                + "    `msg_title` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT "
                + "'消息标题',\n"
                + "    `msg_content` longtext CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '消息内容',\n"
                + "    `msg_content_tags` json NULL COMMENT '消息内容标签',\n"
                + "    `msg_content_type` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT "
                + "'消息内容格式',\n"
                + "    `time_to_live` int(10) UNSIGNED NOT NULL DEFAULT 0 COMMENT '消息过期时间(单位秒; 0-永不过期)',\n"
                + "    `generate_time` datetime NULL DEFAULT NULL COMMENT '业务端产生消息的时间',\n"
                + "    `receive_time` datetime NOT NULL COMMENT '消息中心接收到消息的时间',\n"
                + "    `save_time` datetime NOT NULL COMMENT '消息中心保存消息的时间',\n"
                + "    `expect_exec_time` json NULL COMMENT '预期消息中心消费消息的时间',\n"
                + "    `state` tinyint(2) NOT NULL DEFAULT 0 COMMENT "
                + "'消息状态（负数表示执行前状态，正数表示执行后状态：-3-草稿态；-2-已失效；-1-已过期；0-待执行；1-执行成功；2-执行失败；3-执行中）',\n"
                + "    `status` tinyint(1) NOT NULL DEFAULT 1 COMMENT '数据状态(0-无效；1-有效)',\n"
                + "    `created_by` bigint(20) NULL DEFAULT NULL COMMENT '创建人',\n"
                + "    `created_at` datetime NULL DEFAULT NULL COMMENT '创建时间',\n"
                + "    `updated_by` bigint(20) NULL DEFAULT NULL COMMENT '修改人',\n"
                + "    `updated_at` datetime NULL DEFAULT NULL COMMENT '修改时间',\n"
                + "    PRIMARY KEY (`id`) USING BTREE,\n"
                + "    INDEX `root_id_idx`(`root_id` ASC) USING BTREE,\n"
                + "    INDEX `parent_id_idx`(`parent_id` ASC) USING BTREE,\n"
                + "    INDEX `msg_type_idx`(`msg_type` ASC) USING BTREE,\n"
                + "    INDEX `time_to_live_idx`(`time_to_live` ASC) USING BTREE,\n"
                + "    INDEX `state_idx`(`state` ASC) USING BTREE,\n"
                + "    INDEX `status_idx`(`status` ASC) USING BTREE,\n"
                + "    INDEX `tenant_idx`(`tenant` ASC) USING BTREE\n"
                + ") ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = '消息' ROW_FORMAT = DYNAMIC;");
    }
    
    @Override
    public void initRealTableName(String logicTableName, Set<String> realTableNameSet) {
        if (CollectionUtils.isEmpty(realTableNameSet)) {
            return;
        }
        LOGIC_REAL_TABLE_NAME_CACHE.put(logicTableName, realTableNameSet);
        log.info("initRealTableName -> {}", JSON.toJSONString(LOGIC_REAL_TABLE_NAME_CACHE));
    }
    
    @NonNull
    @Override
    public Set<String> getRealTableName(DatasourceArgs datasourceArgs, String logicTableName) {
        return LOGIC_REAL_TABLE_NAME_CACHE.computeIfAbsent(logicTableName, k -> new HashSet<>());
    }
    
    @NonNull
    @Override
    public Set<String> addRealTableName(@NonNull UpdateActualDataNodeBO bo) {
        String logicTableName = bo.getLogicTableName();
        return batchAddRealTableName(Lists.newArrayList(bo)).get(logicTableName);
    }
    
    @NonNull
    @Override
    public Map<String, Set<String>> batchAddRealTableName(@NonNull List<UpdateActualDataNodeBO> boList) {
        Map<String, Set<String>> resultMap = new HashMap<>(16);
        if (CollectionUtils.isEmpty(boList)) {
            return resultMap;
        }
        /*
         * 判断数据库中的已有表节点是否已经全部包含了本次需要的; 没有包含的话，则将补全所有的节点（数据库中已有的 + 本次需要创建的）到realTableNameSet中，并将当前bo添加到待刷新列表中
         */
        List<UpdateActualDataNodeBO> toRefreshBoList = new ArrayList<>(16);
        for (UpdateActualDataNodeBO bo : boList) {
            String logicTableName = bo.getLogicTableName();
            List<Pair<DatasourceArgs, TreeSet<String>>> dsAndRealTablesPairList = bo.getDsAndRealTablesPairList();
            // 此方法只支持单库分表
            Assert.isTrue(dsAndRealTablesPairList.size() == 1, "curr only support single datasource. curr bo -> " + bo);
            Pair<DatasourceArgs, TreeSet<String>> pair = dsAndRealTablesPairList.get(0);
            Set<String> realTableNameSet = pair.getRight();
            final Set<String> existRealTableNameSet = LOGIC_REAL_TABLE_NAME_CACHE.computeIfAbsent(logicTableName, k -> new HashSet<>());
            if (!existRealTableNameSet.containsAll(realTableNameSet)) {
                // 补全到realTableNameSet中去，因为realTableNameSet会继续往后传递
                realTableNameSet.addAll(existRealTableNameSet);
                // 补全到LOGIC_REAL_TABLE_NAME_CACHE中去
                existRealTableNameSet.addAll(realTableNameSet);
                toRefreshBoList.add(bo);
            }
        }
        
        // 刷新节点
        if (!CollectionUtils.isEmpty(toRefreshBoList)) {
            refreshActualDataNodes(toRefreshBoList);
        }
        
        // 组装返回信息
        resultMap = boList.stream().collect(Collectors.toMap(UpdateActualDataNodeBO::getLogicTableName,
                        x -> x.getDsAndRealTablesPairList().get(0).getRight()));
        return resultMap;
    }
    
    /**
     * 获取真实表的ddl
     *
     * @param logicTableName
     *            逻辑表名
     * @param realTableName
     *            真实表名
     *
     * @return 真实表的ddl
     */
    @NonNull
    @Override
    public String obtainRealTableDdl(DatasourceArgs datasourceArgs, String logicTableName, String realTableName) {
        if (StringUtils.isBlank(logicTableName)) {
            throw new IllegalArgumentException("logicTableName cannot be blank.");
        }
        if (StringUtils.isBlank(realTableName)) {
            throw new IllegalArgumentException("realTableName cannot be blank.");
        }
        if (!TABLE_DDL_MAP.containsKey(logicTableName)) {
            throw new IllegalStateException("Cannot obtain '" + logicTableName + "'s ddl.");
        }
        return String.format(TABLE_DDL_MAP.get(logicTableName), realTableName);
    }
    
    /**
     * 启动的时候刷新所有节点
     */
    @Override
    public void afterSingletonsInstantiated() {
        Set<String> doneTablesSet = new HashSet<>();
        List<UpdateActualDataNodeBO> boList = new ArrayList<>(8);
        for (Map.Entry<String, String> entry : TABLE_DDL_MAP.entrySet()) {
            String logicTableName = entry.getKey();
            if (doneTablesSet.contains(logicTableName)) {
                log.info("logicTable {} has done afterSingletonsInstantiated, skip.", logicTableName);
                continue;
            }
            Set<String> bindingTableSet = rulesShardingProperties.getTableAndBindingTableMap().get(logicTableName);
            if (CollectionUtils.isEmpty(bindingTableSet)) {
                doneTablesSet.add(logicTableName);
                TreeSet<String>  existAllRealTables = DatabaseUtil.obtainTableNameByPrefix(datasourceArgs, logicTableName + "_");
                if (CollectionUtils.isEmpty(existAllRealTables)) {
                    continue;
                }
                // 初始化所有真实表到缓存
                initRealTableName(logicTableName, existAllRealTables);
    
                // 构建UpdateActualDataNodeBO对象
                UpdateActualDataNodeBO bo = UpdateActualDataNodeBO.builder()
                        .logicTableName(logicTableName)
                        .dsAndRealTablesPairList(Lists.newArrayList(Pair.of(datasourceArgs, existAllRealTables)))
                        .build();
                boList.add(bo);
            } else {
                doneTablesSet.addAll(bindingTableSet);
                for (String bindingTable : bindingTableSet) {
                    logicTableName = bindingTable;
                    TreeSet<String>  existAllRealTables = DatabaseUtil.obtainTableNameByPrefix(datasourceArgs, logicTableName + "_");
                    if (CollectionUtils.isEmpty(existAllRealTables)) {
                        continue;
                    }
                    // 初始化所有真实表到缓存
                    initRealTableName(logicTableName, existAllRealTables);
                    // 构建UpdateActualDataNodeBO对象
                    UpdateActualDataNodeBO bo = UpdateActualDataNodeBO.builder()
                            .logicTableName(logicTableName)
                            .dsAndRealTablesPairList(Lists.newArrayList(Pair.of(datasourceArgs, existAllRealTables)))
                            .build();
                    boList.add(bo);
                }
            }
        }
        if (CollectionUtils.isEmpty(boList)) {
            return;
        }
        // 刷新actual-data-nodes
        refreshActualDataNodes(boList);
    }
    
    /**
     * 刷新actual-data-nodes
     */
    private void refreshActualDataNodes(List<UpdateActualDataNodeBO> boList) {
        for (UpdateActualDataNodeBO bo : boList) {
            String logicTableName = bo.getLogicTableName();
            List<Pair<DatasourceArgs, TreeSet<String>>> dsAndRealTablesPairList = bo.getDsAndRealTablesPairList();
            Objects.requireNonNull(dsAndRealTablesPairList, "dsAndRealTablesPairList should not be null.");
            for (Pair<DatasourceArgs, TreeSet<String>> pair : dsAndRealTablesPairList) {
                DatasourceArgs datasourceArgs = pair.getLeft();
                TreeSet<String> realTableNameSet = pair.getRight();
                Objects.requireNonNull(realTableNameSet, "realTableNameSet should not be null.");
                if (CollectionUtils.isEmpty(realTableNameSet)) {
                    realTableNameSet.addAll(DatabaseUtil.obtainTableNameByPrefix(datasourceArgs, logicTableName + "_"));
                }
            }
            dsAndRealTablesPairList = dsAndRealTablesPairList.stream().filter(pair -> !CollectionUtils.isEmpty(pair.getRight())).collect(Collectors.toList());
            bo.setDsAndRealTablesPairList(dsAndRealTablesPairList);
        }
        boList = boList.stream().filter(x -> !CollectionUtils.isEmpty(x.getDsAndRealTablesPairList())).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(boList)) {
            log.warn("boList is empty. skip refresh actual-data-nodes.");
            return;
        }
        refreshActualDataNodesSupport.refresh(boList);
    }
    
}
